package com.example.yckjbigdataflink.transformation;

import com.example.yckjbigdataflink.model.OriginalRow;
import com.example.yckjbigdataflink.model.UserData;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;

// flatMap + 数据清洗与字段标准化: 调用：DataStream<UserData> cleaned = rawStream.flatMap(new DataStandard())
// HBase 原始行数据格式如下：
/*
{
  "rowkey": "abc123",
  "json_data": "{\"uid\":\"user_01\", \"age\":27, \"tags\": [\"vip\", \"beta\"]}"
}
 */
public class DataStandard implements FlatMapFunction<OriginalRow, UserData> {
    private static final ObjectMapper mapper = new ObjectMapper();
    @Override
    public void flatMap(OriginalRow value, Collector<UserData> out) throws Exception {
        if (value.json_data == null) return;

        JsonNode node = mapper.readTree(value.json_data);
        String uid = node.path("uid").asText(null);
        int age = node.path("age").asInt(-1);

        JsonNode tags = node.path("tags");
        if (tags.isArray()) {
            for (JsonNode tagNode : tags) {
                String tag = tagNode.asText();
                if (uid != null && tag != null) {
                    UserData u = new UserData();
                    u.uid = uid;
                    u.age = age;
                    u.tag = tag;
                    out.collect(u);
                }
            }
        }
    }
}
